package com.shujia.hbase.api;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.List;

public class Demo10 {
    private HConnection connection;
    private HBaseAdmin hBaseAdmin;

    @Before
    public void init() {

        //创建hadoop配置文件对象
        Configuration configuration = new Configuration();

        //指定zk地址
        configuration.set("hbase.zookeeper.quorum", "node1:2181,node2:2181,node3:2181");


        try {
            //建立连接   和zk的连接
            connection = HConnectionManager.createConnection(configuration);

            hBaseAdmin = new HBaseAdmin(configuration);

            System.out.println("连接建立成功");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @Test
    public void put() {

        /**
         *create 'dianxin1',{NAME => 'info1',VERSIONS => 10},{NAME => 'info2',VERSIONS => 10000}
         *
         */

        try {
            HTableInterface dianxin = connection.getTable("dianxin1");

            BufferedReader bufferedReader = new BufferedReader(new FileReader("data/dianxin_data"));
            SimpleDateFormat format = new SimpleDateFormat("yyyyMMddHHmmss");
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                String[] split = line.split(",");
                if (!"\\N".equals(split[1])) {


                    String id = split[0];

                    String grid = split[1];

                    String city = split[2];

                    String date = split[7];

                    String time = split[5];


                    Put put = new Put(id.getBytes());


                    /**
                     * 通过手机号查询用户最近10条位置记录
                     *
                     */

                    //保存最近10条数据  一时间中作为版本号
                    long ts = format.parse(time).getTime();
                    put.add("info1".getBytes(), "data".getBytes(), ts, line.getBytes());


                    /**
                     * 获取用户某一天在一个城市中的所有位置
                     *
                     */

                    //以城市编号和日期作为列名
                    String cName = city + "_" + date;

                    put.add("info2".getBytes(), cName.getBytes(), ts, line.getBytes());

                    dianxin.put(put);

                }
            }


        } catch (Exception e) {
            e.printStackTrace();
        }


    }


    @Test
    public void query1() {

        //通过手机号查询用户最近10条位置记录
        try {

            long start = System.currentTimeMillis();
            String id = "609787B15A32690A2B79356127C9006F872D983B";

            HTableInterface dianxin = connection.getTable("dianxin1");

            Get get = new Get(id.getBytes());


            //指定列簇查询
            get.addFamily("info1".getBytes());

            //查询最新10条数据
            get.setMaxVersions(10);


            Result result = dianxin.get(get);

            List<Cell> cells = result.listCells();

            for (Cell cell : cells) {

                String data = Bytes.toString(CellUtil.cloneValue(cell));

                System.out.println(data);
            }


            long end = System.currentTimeMillis();
            System.out.println(end - start);


        } catch (IOException e) {
            e.printStackTrace();
        }

    }


    @Test
    public void query2() {

        //获取用户某一天在一个城市中的所有位置
        try {

            long start = System.currentTimeMillis();


            String id = "7C7F09D9FB218A2F812AFD0DF8C71B1303F17AF0";

            String date = "20180503";

            String city = "83401";


            HTableInterface dianxin = connection.getTable("dianxin1");

            Get get = new Get(id.getBytes());


            //默认只查询最新版本的数据
            get.setMaxVersions(10000);

            String cName = city + "_" + date;

            get.addColumn("info2".getBytes(), cName.getBytes());


            Result result = dianxin.get(get);

            List<Cell> cells = result.listCells();

            for (Cell cell : cells) {

                String data = Bytes.toString(CellUtil.cloneValue(cell));

                System.out.println(data);
            }


            long end = System.currentTimeMillis();
            System.out.println(end - start);


        } catch (IOException e) {
            e.printStackTrace();
        }

    }


}
